package org.example.service;

import io.grpc.ConnectivityState;
import io.grpc.ManagedChannel;
import io.grpc.ManagedChannelBuilder;
import lombok.extern.slf4j.Slf4j;

import java.util.ArrayList;
import java.util.List;
import java.util.Map;
import java.util.concurrent.CopyOnWriteArrayList;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;
import java.util.concurrent.atomic.LongAdder;
import java.util.stream.Collectors;

/**
 * @Author wangjun
 * @Date 2025/3/20 15:06
 */
@Slf4j
public class GrpcClientDemo {

    /**
     * channel列表
     */
    private List<ManagedChannel> channels = new CopyOnWriteArrayList<>();
    /**
     * 可用的channel列表
     */
    private volatile List<ManagedChannel> activeChannels = new ArrayList<>();
    /**
     * 轮询可用channel列表
     */
    private AtomicInteger counter = new AtomicInteger(0);
    /**
     * 定时的线程池,监测可用的channel列表
     */
    private final ScheduledExecutorService scheduler = Executors.newScheduledThreadPool(1);


    public GrpcClientDemo(Map<String, Integer> serverMap) {
        initializeChannels(serverMap);
        scheduleStateUpdate();
    }

    /**
     * 初始化channel
     *
     * @param serverMap
     */
    private void initializeChannels(Map<String, Integer> serverMap) {
        try {
            serverMap.forEach((ip, port) -> {
                ManagedChannel build = ManagedChannelBuilder.forAddress(ip, port).usePlaintext()
                        .defaultLoadBalancingPolicy("round_robin")
                        .enableRetry()
                        .maxRetryAttempts(5)
                        .build();
//                if (ConnectivityState.READY != build.getState(true)) {
//
//                }
                channels.add(build);
            });
        } catch (Exception e) {
            throw new IllegalStateException("Failed to initialize gRPC channels", e);
        }
    }

    // 用于监测通道是否可用
    private void scheduleStateUpdate() {
        scheduler.scheduleAtFixedRate(() -> {
            System.out.println("channel监测");
            List<ManagedChannel> readyChannels = new ArrayList<>();
            // 仅在通道状态变化时创建新列表
            for (ManagedChannel channel : channels) {
                if (channel.getState(true) == ConnectivityState.READY) {
                    readyChannels.add(channel);
                }
            }
            if (!readyChannels.equals(activeChannels)) { // 比较内容是否变化
                activeChannels = new ArrayList<>(readyChannels); // 按需创建新对象
            }
        }, 0, 10, TimeUnit.SECONDS);

        //            List<ManagedChannel> readyChannels = channels.stream()
//                    .filter(channel -> (channel.getState(false) == ConnectivityState.READY || channel.getState(false) == ConnectivityState.IDLE))
//                    .collect(Collectors.toList());
    }

    public ManagedChannel next() {
        List<ManagedChannel> active = activeChannels;
        log.info("当前active的数量：{}", active.size());
        if (active.isEmpty()) {
            throw new IllegalStateException("All channels are inactive");
        }
        int index = counter.getAndUpdate(i -> (i + 1) % active.size());
        System.out.println(index);
        return active.get(index);
    }
}
